#include "jobs.h"

#include "refc_ptr.h"
#include "functional_ptr.h"
#include "util.h"

#include <vector>
#include <queue>
#include <list>
#include <iostream>

using namespace krryn::core;

std::string job::get_name(){
	return typeid(*this).name();
}

class job_group::impl{
public:
	typedef std::vector<job*> job_list_t;
	typedef std::vector<job_group*> simple_deps_list_t;

	job_list_t m_Jobs;
	simple_deps_list_t m_SimpleDeps;
	size_t m_CompletedCounter;
	mutex m_Guard;

	impl() : m_CompletedCounter(0) {}

	~impl(){
		// this lock prevents job_group::impl
		// to be deleted before any processing 
		// (eg. notify_job_done) has completed
		locker<mutex> l_Section(m_Guard);
	}

	void notify_job_done(job_manager &a_Manager){
		locker<mutex> l_Section(m_Guard);

		m_CompletedCounter++;

		// todo the 'any' dependencies

		if(m_CompletedCounter == m_Jobs.size()){
			simple_deps_list_t::iterator i;
			size_t j = 0;
			for(i = m_SimpleDeps.begin(); i != m_SimpleDeps.end(); i++){
				a_Manager.add_group(*i);
				j++;
			}

			m_SimpleDeps.clear();
		}
	}
};

class job_group::group_job : public job{
public:
	group_job(job_group::impl* a_Impl, job_manager &a_Manager, job* a_Job) 
		: m_Job(a_Job), m_Impl(a_Impl), m_Manager(a_Manager){
	}

	virtual void execute(){
		m_Job->execute();
	}

	virtual void done(){
		// The implementation of this function has a couple of problems:
		//
		// 1. Call notify first, before done() is executed because
		//    m_Job->done() could execute an event that might
		//    circumstantially delete m_Impl which means
		//    that we'd be calling a method on deleted memory.
		//
		//    One such event might be an 'end of frame' event with:
		//    
		//    1. The group allocated on the stack
		//    2. The group allocated at the beginning of the frame
		//    3. The wait near the end of the frame
		//
		//    Calling m_Job->done() before notify_job_done() might then
		//    result in m_Imp being deleted because the group is removed
		//    from the stack before the call to notify_job_done() has 
		//    finished. (Which is obviously catastrophic if the function 
		//    accesses it's members, which we assume it does). 
		//
		// 2. However, if we call notify_job_done first before we 
		//    call done() we might be executing code that assumes
		//    the job has completely finished executing even though
		//    it is concurrently executing the done() function.
		//
		// The solution is to aquire a lock to m_Imp in this function,
		// and having the destructor of m_Imp wait for that lock.
		//
		// The only problem here is that notify_job_done() doesn't get
		// called when done() throws an exception.

		try{
			locker<mutex> l_Section(m_Impl->m_Guard);
			m_Job->done();
			m_Impl->notify_job_done(m_Manager);
		}catch(...){
			throw std::runtime_error(
				m_Job->get_name() +
				"::done() failed, dependencies won't be called");
		}
	}

	virtual std::string get_name(){
		return job::get_name() + " (" + m_Job->get_name() + ")";
	}

	virtual ~group_job(){
		delete m_Job;
		m_Job = 0;
	}

	job* m_Job;
	job_group::impl* m_Impl;
	job_manager &m_Manager;
};

job_group::job_group(const std::string &a_Name) : m_Name(a_Name){
	m_Pimpl = 0;
	m_Pimpl = new impl();
}

job_group::~job_group(){
	delete m_Pimpl;
	m_Pimpl = 0;
}

void job_group::add(job* a_Job){
	m_Pimpl->m_Jobs.push_back(a_Job);
}

void job_group::add_to(job_manager &a_Manager){
	impl::job_list_t::iterator i;
	for(i = m_Pimpl->m_Jobs.begin(); i != m_Pimpl->m_Jobs.end(); i++){
		a_Manager.add_job(new group_job(m_Pimpl, a_Manager, *i));
	}
}

void job_group::depends_on_all_of(job_group* a_Group){
	a_Group->m_Pimpl->m_SimpleDeps.push_back(this);
}

void job_group::depends_on_any_of(size_t a_Amount, job_group* a_Group){
}

class job_thread::impl{
public:
	impl(job_manager &a_Manager) : m_Id(m_IdCounter++), m_Manager(a_Manager){
		std::clog << "Starting job thread #" << m_Id << std::endl;
	}

	~impl() {}

	void run(){
		do{
			// thread must continue running even when 
			// an exception occours inside of the job
			try{
				if(job* l_Job = m_Manager.next()){

					l_Job->execute();
					l_Job->done();

					// the job has become obselete
					// (this also means that you can't recycle jobs)
					delete l_Job;
					l_Job = 0;
				}
			}catch(const std::exception &a_Exception){
				locker<mutex> l_Section(m_Guard);
				std::cout << "Exception '" 
					<< a_Exception.what() << "'" << std::endl;
			}catch(...){
				locker<mutex> l_Section(m_Guard);
				std::cout << "Unknown exception" << std::endl;
			}
		}while(true);
	}

	job_manager &m_Manager;

	thread_id_t m_Id;
	static mutex m_Guard;
	static thread_id_t m_IdCounter;
};

krryn::core::mutex job_thread::impl::m_Guard = krryn::core::mutex();
job_thread::thread_id_t job_thread::impl::m_IdCounter = 0;

job_thread::job_thread(job_manager &a_Manager){
	m_Pimpl = new impl(a_Manager);
}

job_thread::~job_thread(){
	delete m_Pimpl;
	m_Pimpl = 0;
}

void job_thread::run(){
	m_Pimpl->run();
}

class job_manager::impl{
public:
	impl(job_manager &a_Manager, size_t a_NumberOfThreads) 
		: m_Threads(a_NumberOfThreads)
	{
		std::generate(m_Threads.begin(), m_Threads.end(), 
			ctor1<job_thread>(ref(a_Manager)));

		std::for_each(m_Threads.begin(), m_Threads.end(), 
			mem_fun_ptr<job_thread*>(&job_thread::start));
	}

	~impl(){
		std::for_each(m_Threads.begin(), m_Threads.end(), 
			mem_fun_ptr<job_thread*>(&job_thread::kill));

		thread_list_t::iterator i = m_Threads.begin();
		for(; i != m_Threads.end(); i++){
			delete *i;
		}
	}

	typedef std::queue<job*> job_list_t;
	typedef std::list<job_thread*> thread_list_t;

	job_list_t m_Jobs;
	thread_list_t m_Threads;
	mutex m_Guard;
};

job_manager::job_manager(size_t a_NumberOfThreads){
	m_Pimpl = 0;
	m_Pimpl = new impl(*this, a_NumberOfThreads);
}

job_manager::~job_manager(){
	delete m_Pimpl;
	m_Pimpl = 0;
}

long job_manager::get_number_of_threads() const{
	return m_Pimpl->m_Threads.size();
}

void job_manager::add_job(job* a_Job){
	locker<mutex> l_Section(m_Pimpl->m_Guard);

	m_Pimpl->m_Jobs.push(a_Job);
}

void job_manager::add_group(job_group* a_Group){
	locker<mutex> l_Section(m_Pimpl->m_Guard);

	a_Group->add_to(*this);
}

job* job_manager::next(){
	// it might happen that m_Pimpl is not initialized yet 
	// while this function is called
	if(m_Pimpl == 0)
		return 0;

	locker<mutex> l_Section(m_Pimpl->m_Guard);

	if(!m_Pimpl->m_Jobs.empty()){
		job* l_Job = m_Pimpl->m_Jobs.front();
		m_Pimpl->m_Jobs.pop();
		return l_Job;
	}

	return 0;
}